package org.korosoft.rusalad.syncnrun.messaging;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Syncnrun messaging route.
 *
 * @author Dmitry Korotkov
 * @since 1.0
 */
public class MessageRoute {
    public static final Log log = LogFactory.getLog(MessageRoute.class);
    public static final int CAPACITY = 100;
    private final UUID name;
    private final BlockingQueue<Message> incoming = new ArrayBlockingQueue<Message>(CAPACITY);
    private final MessageProcessor sender;
    private final AtomicReference<MessageProcessor> hotReceiver = new AtomicReference<MessageProcessor>();

    public MessageRoute(UUID name, MessageProcessor sender) {
        this.name = name;
        this.sender = sender;
    }

    public UUID getName() {
        return name;
    }

    public void put(Message message) throws InterruptedException {
        sender.processMessage(this, message);
    }

    public void setHotReceiver(MessageProcessor receiver) {
        hotReceiver.set(receiver);
        if (receiver != null) {
            for (Message msg = incoming.poll(); msg != null; msg = incoming.poll()) {
                receiver.processMessage(this, msg);
            }
        }
    }

    public MessageProcessor getHotReceiver() {
        return hotReceiver.get();
    }

    public void processIncoming(Message message) throws InterruptedException {
        final MessageProcessor mp = hotReceiver.get();
        if (mp != null) {
            mp.processMessage(this, message);
        } else {
            incoming.put(message);
            if (incoming.size() >= CAPACITY) {
                log.info("Incoming queue " + name + " has hit its capacity");
            }
        }
    }
}
